package com.cupdata.sms.config;

import com.cupdata.sms.core.CusKafkaProperties;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.ByteArrayDeserializer;
import org.apache.kafka.common.serialization.ByteArraySerializer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.Map;
import java.util.UUID;

/**
 * @author: zda
 * @description: TODO
 * @date: 2024/4/29 18:39
 */
@Configuration
public class CoreAutoConfiguration {
    @Bean
    @ConditionalOnProperty(prefix = "channel-core",value = "enableDefaultKafka",matchIfMissing = true,havingValue = "true")
    public KafkaProducer<String,byte[]> defaultProducer(ConfigProperties configProperties){
        Map<String, Object> producerProps = configProperties.getProducerProps();
        if(producerProps.get(ProducerConfig.RETRIES_CONFIG)==null) producerProps.put(ProducerConfig.RETRIES_CONFIG,  3);
        if(producerProps.get(ProducerConfig.BATCH_SIZE_CONFIG)==null) producerProps.put(ProducerConfig.BATCH_SIZE_CONFIG,  16384*2 );
        if(producerProps.get(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG)==null) producerProps.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG,  true );
        if(configProperties.getClientId()==null){
            throw new RuntimeException("必须定义client_id");
        }
        if( configProperties.getBoostrapServers()==null){
            throw new RuntimeException("必须定义bootstrapServers");
        }
        producerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,configProperties.getBoostrapServers());
        return new KafkaProducer(producerProps, new StringSerializer(),new ByteArraySerializer());
    }

}
